package com.learn.transform;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;

/**
 * @create: 2023-04-18 11:06
 * @author: Mr.Du
 * --------------
 * 输入数据为：
 * 上海 浦东新区 777
 * 上海 闵行区 999
 * 上海 杨浦区 666
 * 北京 东城区 567
 * 北京 西城区 987
 * 上海 静安区 888
 * 北京 海淀区 9999
 * @notes:
 **/
public class MinMaxByDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("node1", 9999);

        KeyedStream<Tuple3<String, String, Integer>, Tuple> keyedStream = lines.map(new MapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public Tuple3<String, String, Integer> map(String line) throws Exception {

                return Tuple3.of(line.split(" ")[0], line.split(" ")[1], Integer.parseInt(line.split(" ")[2]));
            }
        }).keyBy(0);


        keyedStream.minBy(2).print("minBy算子最小数据为>>>");
        keyedStream.maxBy(2).print("maxBy算子最大数据为>>>");


        keyedStream.min(2).print("min算子最小数据为>>>");
        keyedStream.max(2).print("max算子最大数据为>>>");

        env.execute("");
    }
}
